package com.study.flink.scala.day02_xcy

import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
import org.apache.flink.api.common.typeinfo.TypeInformation

class KafkaMessageSchemaScala extends  DeserializationSchema[String] with SerializationSchema[String]{
  override def deserialize(message: Array[Byte]): String = {
    new String(message)
  }

  override def isEndOfStream(nextElement: String): Boolean = {
    return false
  }

  override def serialize(element: String): Array[Byte] = {
    element.toString.getBytes
  }

  override def getProducedType: TypeInformation[String] = {
    TypeInformation.of(classOf[String])
  }
}
